package com.aotain.project.apollo.topo;

import org.apache.storm.Config;
import org.apache.storm.StormSubmitter;
import org.apache.storm.kafka.KafkaSpout;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.utils.Utils;

import com.aotain.common.ZkKafkaConfig;
import com.aotain.kafka.KafkaConfigUtil;
import com.aotain.kafka.KafkaProperties;
import com.aotain.project.apollo.bolt.PortDetectBolt;

public class PortDetectTopo {

	public static void main(String[] args) throws Exception {
		TopologyBuilder builder = new TopologyBuilder();
		
		//
		
		ZkKafkaConfig z_kConfig = new ZkKafkaConfig("/home/storm/config/dbconfig.ini");
		
//		ApolloConfig ap = new ApolloConfig("../config/dbconfig.ini");
//		String servername = ap.getMushroomServer();
		
		builder.setSpout("sourceSpout", new KafkaSpout(new KafkaConfigUtil(z_kConfig.getZkConnServer(),KafkaProperties.Abnormal_Log_topic,
				KafkaProperties.zkRoot,KafkaProperties.Log_PortDetect_groupId).getConfig()),10);
		
		
		builder.setBolt("portdetect", new PortDetectBolt()).shuffleGrouping("sourceSpout");
		
		//builder.setBolt("print", new PrintBolt()).shuffleGrouping("exclaim");

		Config conf = new Config();
		conf.setDebug(false);
		conf.setNumWorkers(8);
		conf.setNumAckers(0);
		
		StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
		Utils.sleep(60000);
	}
}
